package rabbitmq_demo

import (
	"context"

	"gitee.com/lihaiping1603/utils/log"

	amqp "github.com/rabbitmq/amqp091-go"
)

//消息确认回调函数:seq:消息的序列号，state:true(成功)，false(失败)
type ConfirmHandler interface {
	AmqpConfirmHandler(seq uint64, state bool)
}

type Producer struct {
	*client
	exchange       string
	confirmHandler ConfirmHandler
	confirmCh      chan amqp.Confirmation
	exit           chan struct{}
}

func NewProducer(confirmHandler ConfirmHandler) *Producer {
	return &Producer{
		client:         &client{},
		exit:           make(chan struct{}),
		confirmHandler: confirmHandler,
	}
}

func (p *Producer) Create(url, exchange, kind string) error {
	p.exchange = exchange
	return p.create(url, exchange, kind)
}

func (p *Producer) Destroy() error {
	close(p.exit)
	return p.destroy()
}

func (p *Producer) StartPublish() error {
	channel := p.channel
	log.Info("enabling publisher confirms.")
	if err := channel.Confirm(false); err != nil {
		log.Error("AMQP channel could not be put into confirm mode: %s", err)
		return err
	}
	confirmCh := channel.NotifyPublish(make(chan amqp.Confirmation, 1))
	p.confirmCh = confirmCh
	go func() {
		p.handleConfirm()
	}()
	return nil
}

func (p *Producer) Publish(ctx context.Context, route_key string, msg amqp.Publishing) (uint64, error) {
	channel := p.channel
	exchange := p.exchange
	seqNo := channel.GetNextPublishSeqNo()

	if err := channel.PublishWithContext(
		ctx,
		exchange,  // publish to an exchange
		route_key, // routing to 0 or more queues
		false,     // mandatory
		false,     // immediate
		msg,
	); err != nil {
		log.Error("AMQP exchange:%s Publish: %s", exchange, err)
		return seqNo, err
	}
	return seqNo, nil
}

func (p *Producer) handleConfirm() {
	for {
		select {
		case <-p.exit:
			log.Info("close the producer")
			return
		case confirmed := <-p.confirmCh:
			seqNo := confirmed.DeliveryTag
			state := confirmed.Ack
			log.Info("confirmed delivery with delivery tag: %d,state:%t", seqNo, state)
			if p.confirmHandler != nil {
				p.confirmHandler.AmqpConfirmHandler(seqNo, state)
			}
		}
	}
}
